package com.apps.sdses.flink141.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/6/13 0:00
 */
public class Kafka2UpsertKafka_PvUv {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        //Source
        //{"user_id": 10001,"page_id": 101,"viewtime": "2022-01-01 12:00:01","user_region": "North America"}
        //{"user_id": 10002,"page_id": 102,"viewtime": "2022-01-01 12:05:23","user_region": "Europe"}
        //{"user_id": 10003,"page_id": 103,"viewtime": "2022-01-01 12:09:17","user_region": "Asia"}
        //{"user_id": 10001,"page_id": 104,"viewtime": "2022-01-01 12:16:59","user_region": "South America"}
        //{"user_id": 10001,"page_id": 104,"viewtime": "2022-01-01 12:16:59","user_region": "South America"}
        //{"user_id": 10002,"page_id": 101,"viewtime": "2022-01-01 12:16:59","user_region": "South America"}
        //{"user_id": 10002,"page_id": 102,"viewtime": "2022-01-01 12:16:59","user_region": "South America"}
        tableEnv.executeSql("CREATE TABLE pageviews (\n" +
                "  user_id BIGINT,\n" +
                "  page_id BIGINT,\n" +
                "  viewtime TIMESTAMP(3),\n" +
                "  user_region STRING,\n" +
                "  WATERMARK FOR viewtime AS viewtime - INTERVAL '2' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'pageviews',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'format' = 'json'\n" +
                ")");

        //Sink
        tableEnv.executeSql("CREATE TABLE pageviews_per_region (\n" +
                "  user_region STRING,\n" +
                "  pv BIGINT,\n" +
                "  uv BIGINT,\n" +
                "  PRIMARY KEY (user_region) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = 'pageviews_per_region',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")");

        //Insert
        tableEnv.executeSql("INSERT INTO pageviews_per_region\n" +
                "SELECT\n" +
                "  user_region,\n" +
                "  COUNT(*),\n" +
                "  COUNT(DISTINCT user_id)\n" +
                "FROM pageviews\n" +
                "GROUP BY user_region");

    }
}
